package com.xzc.apitest.transform

import com.xzc.apitest.source.SensorReading
import org.apache.flink.api.common.functions.{FilterFunction, MapFunction, ReduceFunction, RichMapFunction}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object TransformTest {
  def main(args: Array[String]): Unit = {
    //创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //2-文件
    val stream2 = env.readTextFile("D:\\git\\learning_flink\\_01_试用\\src\\main\\resources\\sensor.txt")
    //    stream2.print()

    val dataStream = stream2
      .map(data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
        //      }).filter(_.id.equals("_1"))
      }).filter(new MyFilter)

    //只取温度最小值
    val aggStream = dataStream
      .keyBy("id")
      //      .min("temperature") //只管温度字段，其它的字段会延用
      .minBy("temperature") //其它字段也是温度最小的那一条的数据

    val reduceStream = dataStream
      .keyBy("id")
      .reduce((curState, newData) =>
        //取温度最小值，以及最新那条数据的时间
        //        SensorReading(curState.id, newData.timestamp, curState.temperature.min(newData.temperature)))
        //取温度最小值，以及最近的时间，这个时间是最近的时间
        SensorReading(curState.id,
          curState.timestamp.max(newData.timestamp),
          curState.temperature.min(newData.temperature)))

    //    aggStream.print()
    //    reduceStream.print()

    //分流操作，打标签
    val splitStream = dataStream
      .split(data => {
        if (data.temperature > 30.0) Seq("high") else Seq("low")
      })
    val highTempStream = splitStream.select("high")
    val lowTempStream = splitStream.select("low")
    val allTempStream = splitStream.select("high", "low")

    //    highTempStream.print("high")
    //    lowTempStream.print("low")
    //    allTempStream.print("all")

    //合流操作 connect 假合-联合国
    val warningStream = highTempStream.map(data => (data.id, data.temperature))
    val connectedStreams = warningStream.connect(lowTempStream)

    //合流操作，coMap对流进行分别处理 高温告警 这就是多维告警
    val coMapResultStream = connectedStreams
      .map(
        warningData => (warningData._1, warningData._2, "warning"),
        lowTempData => (lowTempData.id, "healthy")
      )

    //合流，真合，类型要一样，中国
    val unionStream = highTempStream.union(lowTempStream, allTempStream)

    coMapResultStream.print("coMap")

    //执行
    env.execute("transform test")
  }
}

class MyReduceFunction extends ReduceFunction[SensorReading] {
  override def reduce(t: SensorReading, t1: SensorReading): SensorReading =
    SensorReading(t.id, t1.timestamp, t.temperature.min(t1.temperature))
}

class MyFilter extends FilterFunction[SensorReading] {
  override def filter(t: SensorReading): Boolean =
    t.id.startsWith("_1")
}

class MyMapper extends MapFunction[SensorReading, String] {
  override def map(t: SensorReading): String = t.id + " temperature"
}

class MyRichMapper extends RichMapFunction[SensorReading, String] {
  override def map(t: SensorReading): String = t.id + " temperature"

  //数据来之前，初始化完毕时调用一次
  override def open(parameters: Configuration): Unit = {
    //jdbc get connection
    //获取上下文
    getRuntimeContext
  }

  //数据传输完毕，关闭前-调用一次
  override def close(): Unit = {

  }
}
